package com.huonan.component.es;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.Header;
import cn.hutool.http.HttpRequest;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.huonan.component.es.entity.EsHits;
import com.huonan.component.es.refactoring.ExportArgs;
import com.huonan.utils.ExecutorUtil;
import com.huonan.utils.RetryUtil;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * ES组件，根据输入DSL ， index 进行查询
 */
@Slf4j
@Component
public class ESComponent {

    private static final int FILE_COUNT = 50 * 10000;
    private static String FORMAT_INDEX = "ik_sl_v2_%s_";

    private static final int nThreads = 10;

    /**
     * 输出文件名是否包含索引，小索引（每个索引数据量小）可以不包含索引
     * true： 包含索引
     */
    public static final boolean fileNameIndex = true;
    //环境，prod ,test
    final static ESConfig env;

    static {
        env = EsArgs.initEnv(SystemEnum.PROD);
    }

    /**
     * @param filePath      导出文件，支持文件和目录
     * @param platform      平台
     * @param dsl           dsl语句
     * @param isThread      是否多线程查询
     * @param isSingleIndex 是否全索引，true,只有唯一索引，如用户，快照，榜单等数据源
     * @Description 多个索引，单线程阻塞查询
     * 比如 Index1，Index2
     */
    public static void searchWithThread(String filePath, String platform, String dsl,
                                        IndexDateType indexDateType,
                                        boolean isThread, boolean isSingleIndex) {
        if (isSingleIndex) {
            String index = platform;
            searchWithSingleIndex(filePath, index, dsl);
        } else {
            searchWithAllIndex(filePath, platform, dsl, indexDateType, isThread);
        }
    }

    public static void searchWithThread2(ExportArgs exportArgs) {
        if (exportArgs.isSingleIndex()) {
            String index = exportArgs.getPlatform();
            searchWithSingleIndex(exportArgs.getFilePath(), index, exportArgs.getDsl());
        } else {
            searchWithAllIndex(exportArgs.getFilePath(), exportArgs.getPlatform(), exportArgs.getDsl(),
                    exportArgs.getIndexDateType(),
                    exportArgs.isThread());
        }
    }

    private static void searchWithAllIndex(String filePath, String platform, String dsl, IndexDateType indexDateType, boolean isThread) {
        List<String> indexList;
        FORMAT_INDEX = FORMAT_INDEX + platform;
        indexList = getAllIndex(indexDateType);
        if (isThread) {
            searchWithThread(filePath, indexList, dsl, indexDateType);
        } else {
            for (String nIndex : indexList) {
                searchAsStream(filePath, nIndex, dsl);
            }
        }
    }

    /**
     * 构建全部索引
     * 默认获取30个月索引
     *
     * @param indexDateType 索引类型枚举
     * @return list<全名所以>
     */
    private static List<String> getAllIndex(IndexDateType indexDateType) {
        return get30Month(indexDateType);
    }

    /**
     * 可以单独查询唯一索引，比如：用户，快照等
     *
     * @param filePath 导出文件路径
     * @param index    索引
     * @param dsl      dsl语句
     */
    private static void searchWithSingleIndex(String filePath, String index, String dsl) {
        searchAsStream(filePath, index, dsl);
    }

    private static void searchWithThread(String filePath, List<String> list, String dsl, IndexDateType indexDateType) {
        ExecutorService executor = getExecutorService();
        for (String nIndex : list) {
            String newFilePath = combineNewFilePath(filePath, nIndex, indexDateType);
            CompletableFuture.runAsync(() -> {
                try {
                    searchAsStream(newFilePath, nIndex, dsl);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }, executor);
//            try {
//                searchAsStream(newFilePath, nIndex, dsl);
//            } catch (Exception e) {
//                throw new RuntimeException(e);
//            }
        }
        ExecutorUtil.closePool(executor);
    }

    private static void searchAsStream(String filePath, String index, String dsl) {
        String scroll = null;
        EsHits parse;
        int count = 0;
        String prefix = getFileNamePrefix(filePath, index);
        String fileName = prefix + 1; // 第一个文件名
        while (true) {
            String result = searchByScroll(index, dsl, scroll);
            try {
                parse = ESComponent.parse(result);
                JSONArray hitsObj = parse.getHits();
                if (JSONUtil.isNull(hitsObj) || hitsObj.size() == 0) {
                    break;
                }
                int size = hitsObj.size();
                count += size;
                log.info("索引:{},需导出数据量:{},已导出数据量:{}", index, parse.getHistTotal(), count);
                scroll = parse.getScrollId();
                List<String> list = defaultExtractSource(hitsObj);
                fileName = exportToFile(filePath, count, prefix, fileName, list);
            } catch (Exception e) {
                log.error(e.getMessage());
            }
        }
    }

    private static String combineNewFilePath(String filePath, String input, IndexDateType indexDateType) {

        String patternString = ".*_(\\d{4})(\\d{2}).*"; // 匹配任意字符，然后是一个下划线，然后是4个数字（年份），然后是2个数字（月份），然后再是任意字符
        Pattern pattern = Pattern.compile(patternString);
        Matcher matcher = pattern.matcher(input);
        if (matcher.find()) {
            String datePath = "";
            String year = matcher.group(1); // 获取第一个括号内的匹配结果（年份）
            String month = matcher.group(2); // 获取第二个括号内的匹配结果（月份）
            if (indexDateType.equals(IndexDateType.MOTH_TYPE)) {
                datePath = year;
            } else if (indexDateType.equals(IndexDateType.DAY_TYPE)) {
                datePath = year + FileUtil.FILE_SEPARATOR + month;
            }
            return filePath + FileUtil.FILE_SEPARATOR + datePath;
        }
        return null;
    }

    @NotNull
    private static String getFileNamePrefix(String filePath, String index) {
        String FILE_NAME_PREFIX;
        if (fileNameIndex) {
            FILE_NAME_PREFIX = filePath + FileUtil.FILE_SEPARATOR + DateUtil.today() + "-" + index + "-";
        } else {
            FILE_NAME_PREFIX = filePath + FileUtil.FILE_SEPARATOR + DateUtil.today() + "-";
        }
        return FILE_NAME_PREFIX;
    }

    private static String exportToFile(String filePath, int count, String prefix, String fileName, List<String> list) {
        if (FileUtil.isFile(filePath)) {
            FileUtil.appendUtf8Lines(list, filePath);
        } else {
            if ((count - 10000) % FILE_COUNT == 0) {
                fileName = prefix + (count / FILE_COUNT + 1);
            }
            FileUtil.appendUtf8Lines(list, fileName);
        }
        return fileName;
    }


    private static ExecutorService getExecutorService() {
        return  Executors.newFixedThreadPool(nThreads);
    }

    private static String httpBySql(String url, String sql) {
        String body = HttpRequest.post(url).header(Header.AUTHORIZATION, env.getApiKey())
                .header("Content-Type", "application/json").body(sql).
                setReadTimeout(1000 * 300).execute().body();
        return body;
    }

    /**
     * scrollId 滚动查询
     */
    private static String httpScrollId(String url, String scrollId) {
        JSONObject body = new JSONObject();
        body.putOpt("scroll_id", scrollId);
        body.putOpt("scroll", env.getScrollTime());
        HttpRequest.post(url).header(Header.AUTHORIZATION, env.getApiKey())
                .header("Content-Type", "application/json");
        return HttpRequest.post(url).header(Header.AUTHORIZATION, env.getApiKey())
                .header("Content-Type", "application/json")
                .body(body.toString()).timeout(1000 * 300).execute().body();
    }

    public static EsHits parse(String json) throws Exception {
        EsHits esHits = new EsHits();
        JSONObject jsonObj = JSONUtil.parseObj(json);
//        if (checkResult(jsonObj)) {
//            return esHits;
//        }
        String scrollId = jsonObj.getStr("_scroll_id");
        int histTotal = jsonObj.getJSONObject("hits").getJSONObject("total").getInt("value");
        JSONArray jsonArray = jsonObj.getJSONObject("hits").getJSONArray("hits");
        esHits.setHits(jsonArray);
        esHits.setScrollId(scrollId);
        esHits.setHistTotal(histTotal);
        return esHits;
    }

    private static boolean checkResult(JSONObject jsonObj) {
        String str = jsonObj.getStr("error");
        if (StrUtil.isNotEmpty(str)) {
            log.error("查询es报错,es返回报错信息:{}", str);
            return true;
        }
        return false;
    }

    private static List<String> defaultExtractSource(JSONArray hits) {
        List<String> list = new ArrayList<>();
        hits.forEach(s -> {
            JSONConfig jsonConfig = new JSONConfig();
            jsonConfig.setIgnoreNullValue(false);
            JSONObject newObj = new JSONObject(jsonConfig);
            JSONObject object = JSONUtil.parseObj(s);
            String index = object.getStr("_index");
            String id = object.getStr("_id");
            String _type = object.getStr("_type");
            newObj.putOpt("_index", index);
            newObj.putOpt("_id", id);
            newObj.putOpt("_type", _type);
            String source = object.getStr("_source");
            //TODO 默认导出_source 字段
            SourceFields.defaultExtractSource(newObj, source);
//            TODO 自定义导出默认导出_source 字段格式,没指定_source默认导出所有
//            SourceFields.exportSource(newObj, source);
            list.add(JSONUtil.toJsonStr(newObj));
        });
        return list;
    }

    /**
     * 获取agg中聚合的发布时间
     */
    private static List<String> extractAgg(String aggs) {
        JSONObject entries = JSONUtil.parseObj(aggs);
        JSONObject aggregations = entries.getJSONObject("aggregations");
        JSONObject publishedAtMonth = aggregations.getJSONObject("date_publishedAt_month");
        JSONArray buckets = publishedAtMonth.getJSONArray("buckets");
        List<String> list = new ArrayList<>();
        for (Object bucket : buckets) {
            JSONObject object = JSONUtil.parseObj(bucket);
            String month = object.getStr("key_as_string");
            list.add(month);
        }
        return list;
    }

    private static String searchBySql(String index, String dsl) {
        String url = getSqlUrl(env.getUrl(), index);
        return httpBySql(url, dsl);
    }

    private static String searchByScroll(String index, String dsl) {
        String url = getFirstScrollUrl(env.getUrl(), index);
        return httpBySql(url, dsl);
    }

    private static String searchByScroll(String scroll) {
        return httpByScroll(scroll);
    }


    private static String searchByScroll(String index, String dsl, String scroll) {
        RetryUtil instance = RetryUtil.getInstance();
//        instance.retry(() -> {
//            if (StrUtil.isEmpty(scroll)) {
//                return searchByScroll(index, dsl);
//            } else {
//                return searchByScroll(scroll);
//            }
//        }, 3, 5000);
//        return null;
        if (StrUtil.isEmpty(scroll)) {
            return searchByScroll(index, dsl);
        } else {
            return searchByScroll(scroll);
        }
    }

    private static String httpByScroll(String scroll) {
        String url = getScrollUrl(env.getUrl());
        return httpScrollId(url, scroll);
    }

    private static String getSqlUrl(String url, String index) {
        return String.format("%s/%s/%s", url, index, "_search");
    }

    private static String getFirstScrollUrl(String url, String index) {
        return String.format("%s/%s/%s", url, index, "_search?scroll=" + env.getScrollTime());
    }

    private static String getScrollUrl(String url) {
        return String.format("%s/%s", url, "_search/scroll");
    }

    /**
     * 根据索引类型（月、天）获取范围内的全部索引，比如一年内的
     *
     * @param type
     * @return
     */
    private static List<String> get30Month(IndexDateType type) {
        // 时间offset ，比如12代表12二个月内的索引
        int dateOffset = 12;
        //获取指定时间内的所有月份日期
        DateTime offset = DateUtil.offset(new DateTime(), DateField.MONTH, -dateOffset);
        List<String> dateTimes = new ArrayList<>();
        if (type.equals(IndexDateType.MOTH_TYPE)) {
            //获取月粉索引
            dateTimes = DateUtil.rangeToList(offset, new DateTime(), DateField.MONTH)
                    .stream().map(d -> DateUtil.format(d, type.getValue())).collect(Collectors.toList());
        } else if (type.equals(IndexDateType.DAY_TYPE)) {
            //获取天索引
            //获取指定时间范围的月份开始时间
            DateTime monthTimeStart = DateUtil.rangeToList(offset, new DateTime(), DateField.MONTH).get(0);
            //获取指定时间范围的全部日期
//            List<DateTime> dayMonthStart = DateUtil.rangeToList(monthTimeStart, new DateTime(), DateField.DAY_OF_YEAR);
            List<String> dayIndex = DateUtil.rangeToList(monthTimeStart, new DateTime(), DateField.DAY_OF_YEAR)
                    .stream().map(d -> DateUtil.format(d, IndexDateType.DAY_TYPE.getValue()))
                    .collect(Collectors.toList());
//            int monthCount = 30 - dayMonthStart.size();
//            DateTime monthStart = DateUtil.offsetMonth(monthTimeStart, -(monthCount + 1));
//            List<String> monthIndex = DateUtil.rangeToList(monthStart, monthTimeStart, DateField.MONTH)
//                    .stream().map(d -> DateUtil.format(d, IndexDateType.MOTH_TYPE.getValue()))
//                    .collect(Collectors.toList());
//            dateTimes.addAll(monthIndex);
            dateTimes.addAll(dayIndex);
        }
        return dateTimes.stream()
                .map(t -> String.format(FORMAT_INDEX, t))
                .collect(Collectors.toList());
    }
}
